#pragma once

#include "AbstractRequestService.h"
#include "broker/BrokerConfig.h"
#include "base/TokenBucket.h"
#include "RequestDispatcher.h"

namespace zeroplus::mq
{
    class PubService : public AbstractRequestService
    {
    private:
        std::shared_ptr<RedisStore> redisStore_;
        TokenBucket<std::chrono::steady_clock> tokenBucket_;
        RequestDispatcher dispatcher_;

    public:
        PubService(std::shared_ptr<BrokerMediator> mediator)
            : AbstractRequestService(mediator),
              redisStore_(RedisStore::getInstance(BrokerConfig::getInstance().redisAddr))
        {
            dispatcher_.RegisterHandler<transport::RequestCode::PUB_MESSAGE, transport::PublishMessageRequest, transport::PublishMessageResponse>(
                std::bind(&PubService::ProcessPublishMessageRequest, this, std::placeholders::_1));
        }

        ~PubService()
        {
        }

        std::optional<transport::ReplyCommand> ProcessRequest(const transport::RequestCommand &cmd) override
        {
            return dispatcher_.Process(cmd);
        }

        void InternalSend(const Bytes &data) override
        {
            mediator_->Notify(ServiceType::PublishService, data);
        }

        transport::PublishMessageResponse ProcessPublishMessageRequest(const transport::PublishMessageRequest &request)
        {
            LOG_DEBUG("处理消息发布请求");
            std::string topic = request.topic();
            uint64_t delay = request.delay();
            const Bytes msgBytes = Bytes::from_string(request.body());

            if (!redisStore_->IsTopicExists({topic}))
            {
                LOG_ERROR("register topic %s", topic.c_str());
                InternalSend(Bytes::from_string(topic));
            }

            //  一个topic多个队列可以采取创建多个以topic:queueid为格式的redislist
            if (delay > 0)
            {
                LOG_DEBUG("写入延迟队列");
                auto res = redisStore_->WriteToDQ(topic, delay, msgBytes);
            }
            else
            {
                LOG_DEBUG("写入消息队列");
                auto messageId = redisStore_->WriteToMQ(topic, msgBytes);
            }
            // 响应客户端
            transport::PublishMessageResponse response;
            response.set_code(transport::ResponseCode::SUCCESS);
            return response;
        }
    };
}
